package com.ra.component.mq;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.stereotype.Service;

import com.espirit.eap.sa.LoginUser;
import com.espirit.eap.sa.LoginUserService;
import com.googlecode.cswish.exception.ApplicationException;
import com.googlecode.cswish.model.PageInfo;
import com.googlecode.cswish.oosearch.QLInfo;
import com.googlecode.cswish.struts.Initialization;
import com.googlecode.cswish.struts.spring.Executor;
import com.googlecode.cswish.struts.spring.GenericJPAService;
import com.googlecode.cswish.struts.spring.GenericService;
import com.ra.landlord.TenantUser;
import com.ra.util.RaConstant;

@Service
public class MessageDataService implements Initialization{
	
	@Resource
	private GenericService genericService;
	
	@Resource
	private GenericJPAService genericJPAService;
	
	@Resource
	private LoginUserService loginUserService;
	
	private static String MESSAGE_TYPE_GLOBAL_KEY = "MESSAGE_GLOBAL";
	
	private static final Log logger = LogFactory.getLog(MessageDataService.class);
	
	//ConcurrentHashMap线程安全
	private static Map<String,List<MessageData>> messageMap = new ConcurrentHashMap<>(1024);
	
	
	/**
	 * 初始化的时候价值数据库未读的普通消息和一年内的系统消息
	 */
	@Override
	public void initSite() {
		//初始化一年内的系统消息
		initGlobalMessage();
		//初始化未读的点到点消息
		initCommonMessage();
	}
	
	@Override
	public String[] getProductName() {
		return RaConstant.PRODUCT_NAME;
	}
	
	/**
	 * 初始化未读的点到点消息
	 */
	private void initCommonMessage(){
		logger.debug("========= init unread commonMessage.");
		String ql = "select a from " + MessageData.class.getName() + " a where a.status = " + MessageData.MESSAGE_UNREAD + " and a.messageType = " + MessageData.MESSAGE_TYPE_COMMON;
		QLInfo qlInfo = new QLInfo(ql, true);
		List<MessageData> commonMessage = genericService.searchByQl(qlInfo, false).getData();
		if (commonMessage != null && commonMessage.size() > 0) {
			for (MessageData messageData : commonMessage) {
				String key = String.valueOf(messageData.getReceiver().getId());
				List<MessageData> messageDatas = messageMap.get(key);
				if (messageDatas == null) {
					messageDatas = new ArrayList<>();
					messageMap.put(key, messageDatas);
				}
				messageDatas.add(messageData);
			}
			logger.debug("========= init unread commonMessage end. " + commonMessage.size() + " message.");
		}
	}
	
	/**
	 * 初始化一年内的系统消息
	 */
	private void initGlobalMessage(){
		Calendar calendar = Calendar.getInstance();
		calendar.add(Calendar.YEAR, -1);
		String ql = "select a from " + MessageData.class.getName() + " a where a.createTime > ? and a.messageType = " + MessageData.MESSAGE_TYPE_GLOBAL;
		QLInfo qlInfo = new QLInfo(ql, true, calendar.getTime());
		List<MessageData> globalMessageDatas = genericService.searchByQl(qlInfo, false).getData();
		messageMap.put(MESSAGE_TYPE_GLOBAL_KEY, globalMessageDatas);
	}
	
	/**
	 * 发送普通消息
	 * @param message
	 */
	public void send(MessageData message){
		message.setMessageType(MessageData.MESSAGE_TYPE_COMMON);
		sendMessage(message);
	}
	
	/**
	 * 发送系统消息
	 * @param message
	 */
	public void sendGlobalMessage(MessageData message){
		message.setMessageType(MessageData.MESSAGE_TYPE_GLOBAL);
		sendMessage(message);
	}
	
	/**
	 * 发送消息
	 * @param message
	 */
	public void sendMessage(MessageData message){
		
		LoginUser loginUser = loginUserService.getLoginUser();
		message.setSender(loginUser);
		message.setCreateTime(Calendar.getInstance().getTime());
		message.setStatus(MessageData.MESSAGE_UNREAD);
		message.setChannel(MessageData.MESSAGE_CHANNEL_ALL);
		
		//异步线程插入数据库
		genericService.executeAsync(new Executor<MessageData, MessageData>() {
			@Override
			public MessageData execute(MessageData message) {
				MessageData resultMessageData = genericService.add(message);
				String key = null;
				//存入messageMap
				if (resultMessageData.getMessageType() == MessageData.MESSAGE_TYPE_GLOBAL) {
					key = MESSAGE_TYPE_GLOBAL_KEY;
				}else {
					key = String.valueOf(resultMessageData.getReceiver().getId());
				}
				List<MessageData> messageDatas = messageMap.get(key);
				if (messageDatas == null) {
					messageDatas = new ArrayList<>();
					messageMap.put(key, messageDatas);
				}
				messageDatas.add(resultMessageData);
				//logger.debug("========="+message.getSender().getId() + " send a message to "+message.getReceiver().getId());
				return resultMessageData;
			}
		}, message);
	}
	
	/**
	 * 接收普通消息
	 */
	public List<MessageData> receive(){
		
		LoginUser loginUser = loginUserService.getLoginUser();
		
		MessageData messageDataModel = new MessageData();
		messageDataModel.setReceiver(loginUser);
		messageDataModel.setStatus(MessageData.MESSAGE_UNREAD);
		
		//将内存中的当前登录用户的消息全部读出来
		List<MessageData> messageDatas = null;
		if (loginUser != null) {
			String key = String.valueOf(loginUser.getId());
			messageDatas = messageMap.get(key);
			if (messageDatas != null) {
				//异步将数据库中的消息改为已读
				genericService.executeAsync(new Executor<Boolean, List<MessageData>>() {
					@Override
					public Boolean execute(List<MessageData> messageDatas) {
						
						Iterator<MessageData> messageIter = messageDatas.iterator();
						while (messageIter.hasNext()) {
							MessageData messageData = messageIter.next();
							messageData.setStatus(MessageData.MESSAGE_READED);
							genericService.merge(messageData);
						}
						return true;
					}
				}, messageDatas);
				//logger.debug("========="+loginUser.getId()+" receive " + messageDatas.size() + " message");
			}
			messageMap.remove(key);
		}
		
		return messageDatas;
	}

	/**
	 * 接收系统消息（登录时才获取系统消息，不需要轮询获取）
	 * @return
	 */
	public List<MessageData> receiveGlobalMessage(){
		List<MessageData> unreadMessageDatas = new ArrayList<>();
		Map<String,Object> mapMessageInfo = new HashMap<>();
		LoginUser loginUser = loginUserService.getLoginUser();
		
		//获取一年内的系统消息
		List<MessageData> messageDatas = null;
		messageDatas = messageMap.get(MESSAGE_TYPE_GLOBAL_KEY);
		//获取用户已读的消息
		MessageLoginUser messageLoginUserModel = new MessageLoginUser();
		messageLoginUserModel.setReceiver(loginUser);
		messageLoginUserModel.setStatus(MessageData.MESSAGE_READED);
		List<MessageLoginUser> messageLoginUsers = genericService.searchAll(messageLoginUserModel).getData();
		
		//获取到未读的消息
		if (messageDatas != null && messageDatas.size() > 0) {
			int fromIndex = 0;
			if (messageLoginUsers != null && messageLoginUsers.size() > 0) {
				MessageData lastReadedMessageData = messageLoginUsers.get(messageLoginUsers.size() - 1).getMessageData();
				for (int i = messageDatas.size(); i > 0; i--) {
					MessageData messageData = messageDatas.get(i - 1);
					if (messageData.getId().equals(lastReadedMessageData.getId())) {
						fromIndex = i - 1;
					}
				}
			}
			if ((fromIndex + 1) < messageDatas.size()) {
				LoginUser newLoginUser = new LoginUser();
				newLoginUser.setId(loginUser.getId());
				messageDatas.get(messageDatas.size() - 1).setReceiver(newLoginUser);
				unreadMessageDatas = messageDatas.subList(fromIndex, messageDatas.size());
				mapMessageInfo.put("messageData", unreadMessageDatas);
				mapMessageInfo.put("loginUser", loginUserService.getLoginUser());
			}
		}

		//异步将未读的消息持久化数据库，并设置为已读
		genericService.executeAsync(new Executor<Boolean, Map<String,Object>>() {
			@Override
			public Boolean execute(Map<String,Object> mapMessageInfo) {
				LoginUser loginUser = (LoginUser)mapMessageInfo.get("loginUser");
				List<MessageData> messageDatas = (List<MessageData>)mapMessageInfo.get("messageData");
				if (messageDatas != null) {
					for (MessageData messageData : messageDatas) {
						MessageLoginUser newMessageLoginUser = new MessageLoginUser();
						newMessageLoginUser.setMessageData(messageData);
						newMessageLoginUser.setReceiver(loginUser);
						newMessageLoginUser.setStatus(MessageData.MESSAGE_READED);
						newMessageLoginUser.setReadTime(new Date());
						genericService.add(newMessageLoginUser);
					}
				}
				return true;
			}
		}, mapMessageInfo);
		
		return unreadMessageDatas;
	}
	
	/**
	 * 查询已读的消息
	 * @return
	 */
	public List<MessageData> searchReadedMessage(MessageData model, Integer pageIndex, Integer pageSize){
		if (pageIndex == null || pageIndex < 1) {
			pageIndex = 1;
		}
		if (pageSize == null || pageIndex < 1) {
			pageSize = 15;
		}
		String strShort = "id"; 
		String strOrderBy = "DESC";
		LoginUser loginUser = loginUserService.getLoginUser();
		model.setReceiver(loginUser);
		model.setStatus(MessageData.MESSAGE_READED);
		PageInfo<MessageData> pageInfo = this.genericService.search(model, (pageIndex - 1) * pageSize, pageSize, strShort, strOrderBy, false, null, null, true, false);
		return pageInfo.getData();
	}
	
	/**
	 * 删除消息
	 * @param messageDatas
	 * @return
	 */
	public boolean deleteMessageData(List<Object> messageDatas) {
		boolean result = genericJPAService.deleteList(messageDatas);
		if (!result) {
			throw new ApplicationException("删除消息失败");
		}
		return result;
	}
	
	/**
	 * 测试
	 */
	public void testSend() {
		for (int i = 0; i < 100; i++) {
			if (i != 0 && i % 2 == 0) {
				try {
					Thread.sleep(6000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}	
			}
			
			MessageData messageData = new MessageData();
			messageData.setContent("this is messagedata test: "+i);
			messageData.setTitle("title"+i);
			TenantUser tenantuser = genericService.load(TenantUser.class, 7691);
			LoginUser receiver = tenantuser.getLoginUser();
			messageData.setReceiver(receiver);
			messageData.setMessageType(MessageData.MESSAGE_TYPE_COMMON);
			messageData.setChannel(MessageData.MESSAGE_CHANNEL_ALL);
			send(messageData);
		}
	}

}